package chapter14

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.sql.SparkSession

/**
 * author: yuhui
 * descriptions:
 * date: 2024 - 10 - 26 3:50 下午
 *
 * CREATE TABLE view_top(
 * dt char(20) ,
 * url char(50),
 * url_count int
 * )ENGINE=InnoDB DEFAULT CHARSET=utf8;
 *
 */

object DataViewTopN {

  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .appName("")
      .master("local[*]")
      .getOrCreate()

    val lines = spark.sparkContext.textFile("doc/output_buffered.txt")
    //将RDD关联了Schema，但是依然是RDD
    val rddBean = lines.map(line => {
      val fields = line.split(",")
      new DataView(fields(0), fields(1), fields(2))
    })

    val dataDF = spark.createDataFrame(rddBean, classOf[DataView])
    dataDF.show()

    dataDF.createTempView("dataTable")

    /** *
     * 1）WITH RankedUrls AS (...) 定义了一个公用表表达式（CTE），它首先按dt和url分组，并计算每个dt中每个url的出现次数（url_count）。
     * 2）ROW_NUMBER() OVER (PARTITION BY dt ORDER BY COUNT(*) DESC) 为每个dt分组内的url分配一个排名，排名依据是url的出现次数（降序）。
     * 3）在外部查询中，我们从RankedUrls CTE中选择dt、url和url_count，但只选择排名在前五的记录（WHERE rank <= 5）。
     * 4）最后，我们按dt和rank对结果进行排序，以确保输出是有序的。
     */

    val frame = spark.sql(
      """
        |
        | WITH RankedUrls AS (
        |  SELECT
        |    dt,
        |    url,
        |    count(*) as url_count,
        |    ROW_NUMBER() OVER (PARTITION BY dt ORDER BY COUNT(*) DESC) AS rank
        |  FROM
        |    dataTable
        |  GROUP BY
        |    dt,
        |    url
        |)
        |
        |SELECT
        |  dt,
        |  url,
        |  cast(url_count as int) as url_count
        |FROM
        |  RankedUrls
        |WHERE
        |  rank <= 5
        |ORDER BY
        |  dt,
        |  rank;
        |
        |""".stripMargin)

    //5、把数据保存到mysql表中
    frame.rdd.foreach(line => {
      //每条数据与mysql建立连接
      //把数据插入到mysql表操作
      //1、获取连接
      val connection: Connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/ngn", "root", "123456")

      //2、定义插入数据的sql语句
      val sql = "insert into view_top(dt,url,url_count) values(?,?,?)"

      //3、获取PreParedStatement

      try {
        val ps: PreparedStatement = connection.prepareStatement(sql)

        //4、获取数据,给？号 赋值
        ps.setString(1, line.getAs[String](0))
        ps.setString(2, line.getAs[String](1))
        ps.setInt(3, line.getAs[Int](2))
        //执行
        ps.execute()
      } catch {
        case e: Exception => e.printStackTrace()
      } finally {
        if (connection != null) {
          connection.close()
        }
      }
    })

  }
}